package cn.xinfei.xdecision.engine.runner.executor.strategy.branch;

import cn.xinfei.xdecision.common.model.component.strategyout.StrategyOutput;
import cn.xinfei.xdecision.common.model.enginex.po.EngineNode;
import cn.xinfei.xdecision.common.utils.enums.enginex.NodeTypeEnum;
import cn.xinfei.xdecision.common.vo.FieldInfoVo;
import cn.xinfei.xdecision.engine.runner.context.PipelineContext;
import cn.xinfei.xdecision.engine.runner.context.PipelineMetaInfoContext;
import cn.xinfei.xdecision.engine.runner.context.XDecisionContext;
import cn.xinfei.xdecision.engine.runner.executor.NodeHandler;
import cn.xinfei.xdecision.engine.runner.executor.DecisionFlowPipeline;
import cn.xinfei.xdecision.engine.runner.executor.XRunner;
import cn.xinfei.xdecision.engine.runner.executor.model.StateTransmitModel;
import cn.xinfei.xdecision.engine.runner.executor.strategy.EngineNodeRunner;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

/**
 * 并行节点
 */
@XRunner
@Slf4j
@Service
public class Parallel extends EngineNodeRunner {

    @Resource(name = "bizProcessorThreadPool")
    private ExecutorService bizProcessorThreadPool;

    @Override
    public Map<String, StrategyOutput> getTerminalOutputFieldInfo(NodeHandler nodeHandler) {

        Map<String, StrategyOutput> fieldVos = new HashMap<>();

        return fieldVos;
    }
    @Override
    public String getServiceName() {
        return EngineNodeRunner.PRE+ NodeTypeEnum.PARALLEL.getType();
    }
    public Map<String, FieldInfoVo> getInputFieldInfo(NodeHandler nodeHandler) {
        return null;
    }
    @Override
    public Map<String, StrategyOutput> getOutputFieldInfo(NodeHandler nodeHandler) {
        return null;
    }
    /**
     * 并行节点处理（并行执行后面的分支，并返回最后的聚合节点）
     *
     * @param nodeHandler
     */
    @Override
    public StateTransmitModel runNode(NodeHandler nodeHandler) {
        EngineNode engineNode = nodeHandler.getEngineNode();
        String nodeCode = engineNode.getNodeCode();
        XDecisionContext context = XDecisionContext.getContext();
        PipelineContext pipelineContext = context.getPipelineContext();
        log.info("执行并行节点 runNode engineNode:{}", engineNode.toSimpleString());
        String nextNodeCodes = engineNode.getNextNodeCodes();
//        NodeHandler aggregationNode = null; // 聚合节点code
        String aggregationNode = null; // 聚合节点code
        String[] nextNodeArr = nextNodeCodes.split(",");
        List<CompletableFuture<String>> futureList = asyncHandler(nextNodeArr, context);
        for (CompletableFuture<String> future : futureList) {
            try {
                String result = future.get();
                if (result != null) {
                    aggregationNode = result;
                }
            } catch (Exception e) {
                log.error("异步分支执行异常，nodeCode={}", nodeCode, e);
            }
        }
        log.info("执行并行节点 runNode nodeCode:{},nextNode:{}", nodeCode, aggregationNode);

        return  XDecisionContext.getContext().getStateContext().setNextNodeCode(aggregationNode);
    }

    private List<CompletableFuture<String>> asyncHandler(String[] nextNodeArr, XDecisionContext context) {
        List<CompletableFuture<String>> futureList = new ArrayList<>();
        PipelineMetaInfoContext pipelineMetaInfoContext = context.getPipelineMetaInfoContext();
        for (String nextNodeCode : nextNodeArr) {
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                try {

                    NodeHandler nextNodeHandler = pipelineMetaInfoContext.get(nextNodeCode);
                    String nodeCode = nextNodeHandler.getEngineNode().getNodeCode();
                    StateTransmitModel aggregationNodeCode = DecisionFlowPipeline.shoot(nextNodeHandler);
                    log.info("异步分支执行结束，nodeCode={},返回聚合节点nextNodeCode={},startNodeCode={}", nodeCode, aggregationNodeCode, nextNodeCode);
                    return aggregationNodeCode.getNextNodeCode();
                } catch (Exception e) {
                    log.error("异步分支执行异常，nodeCode={}", nextNodeCode, e);
                }
                return null;
            }, bizProcessorThreadPool);
            futureList.add(future);
        }
        return futureList;
    }




    @Override
    public boolean terminalCondition(NodeHandler nodeHandler, Map<String, Object> nodeOutput, Map<String, Object> nodeInput) {
        return false;
    }
}
